package com.atguigu.day01;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

// 从socket消费数据，然后单词计数
public class Example1 {
    // 不要忘记抛出异常！！！
    public static void main(String[] args) throws Exception {
        // 获取流执行环境（程序的上下文）
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 从socket读取数据然后输出
        env
                // 先启动`nc -lk 9999`
                // "hello world"
                .socketTextStream("localhost", 9999)
                // 设置数据源的并行子任务的数量为1
                .setParallelism(1)
                // map阶段
                // "hello world" => ("hello", 1), ("world", 1)
                // 一对多转换
                .flatMap(new Tokenizer())
                .setParallelism(1)
                // shuffle阶段
                // 按照单词对数据流进行分组
                // KeySelector<IN, KEY>
                .keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
                    @Override
                    public String getKey(Tuple2<String, Integer> in) throws Exception {
                        // 将in元组的f0字段指定为in的key，f0相当于scala中的`._1`
                        return in.f0;
                    }
                })
                // reduce阶段
                .reduce(new WordCount())
                .setParallelism(1)
                .print()
                .setParallelism(1);

        // 提交任务
        env.execute();
    }

    // FlatMapFunction<IN, OUT>
    public static class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
        // flatMap的两个参数
        // 第一个参数：输入数据
        // 第二个参数：集合，将要输出的数据收集到集合中，flink将集合中的数据自动发送下去
        @Override
        public void flatMap(String in, Collector<Tuple2<String, Integer>> out) throws Exception {
            String[] words = in.split(" ");
            for (String word : words) {
                // 把要输出的数据收集到集合中，flink将集合中的数据自动发送下去
                out.collect(Tuple2.of(word, 1));
            }
        }
    }

    // ReduceFunction<IN>
    // 因为reduce输入、累加器和输出的泛型是一样的
    public static class WordCount implements ReduceFunction<Tuple2<String, Integer>> {
        // 新累加器 reduce(累加器，输入数据)
        // reduce定义的是输入数据和累加器的聚合规则
        // 将新累加器输出
        @Override
        public Tuple2<String, Integer> reduce(Tuple2<String, Integer> accumulator, Tuple2<String, Integer> in) throws Exception {
            return Tuple2.of(accumulator.f0, accumulator.f1 + in.f1);
        }
    }
}
